package com.five.listener;

import com.alibaba.fastjson.JSON;
import com.five.dto.CollectionDto;
import com.five.message.CollectMessage;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;


@RocketMQTransactionListener(rocketMQTemplateBeanName = "addCollectRocketMQTemplate")
public class CollectTransactionListener implements RocketMQLocalTransactionListener {
    private static final Logger log = LoggerFactory.getLogger(CollectTransactionListener.class);
    private static final String COLLECT_KEY_PREFIX = "reader:collect:";
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {

        byte[] payloadBytes = (byte[]) msg.getPayload();
        CollectMessage msgBody = JSON.parseObject(payloadBytes, CollectMessage.class);

        CollectionDto dto = (CollectionDto) arg;

        Long readerId = dto.getReaderId();
        Long novelId = dto.getNovelId();


        try {
            if (msgBody.getOperateType() == 1) {
                // 新增收藏：写入Redis (Set集合，天然去重）
                String redisKey = COLLECT_KEY_PREFIX + readerId;
                Long addCount = redisTemplate.opsForSet().add(redisKey, String.valueOf(novelId));
                if (addCount == 0) {
                    // 已收藏，无需后续操作（消息直接提交，但消费端会做幂等处理）
                    return RocketMQLocalTransactionState.COMMIT;
                }
                // Redis写入成功，提交消息
                return RocketMQLocalTransactionState.COMMIT;
            } else if (msgBody.getOperateType() == 0) {
                // 取消收藏：删除Redis
                String redisKey = COLLECT_KEY_PREFIX + readerId;
                Long removeCount = redisTemplate.opsForSet().remove(redisKey, String.valueOf(novelId));
                if (removeCount == 0) {
                    // 未收藏，直接提交消息（消费端会做幂等）
                    return RocketMQLocalTransactionState.COMMIT;
                }
                // Redis删除成功，提交消息
                return RocketMQLocalTransactionState.COMMIT;
            }
            return RocketMQLocalTransactionState.ROLLBACK;
        } catch (Exception e) {
            log.error("本地事务失败", e);
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    // 事务回查（解决本地事务执行后，消息状态未知的情况）
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // 解析消息体
        byte[] payloadBytes = (byte[]) msg.getPayload();
        CollectMessage msgBody = JSON.parseObject(payloadBytes, CollectMessage.class);

        Long readerId = msgBody.getReaderId();
        Long novelId = msgBody.getNovelId();

        String redisKey = COLLECT_KEY_PREFIX + readerId;
        Boolean isMember = redisTemplate.opsForSet().isMember(redisKey, String.valueOf(novelId));

        if (msgBody.getOperateType() == 1) { // 新增收藏
            // 本地事务成功：Redis中存在该记录 → 提交消息
            if (Boolean.TRUE.equals(isMember)) {
                return RocketMQLocalTransactionState.COMMIT;
            }
        } else if (msgBody.getOperateType() == 0) { // 取消收藏
            // 本地事务成功：Redis中不存在该记录 → 提交消息
            if (Boolean.FALSE.equals(isMember)) {
                return RocketMQLocalTransactionState.COMMIT;
            }
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }
}